package org.gorson.executor;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.gorson.BO.SensorReadingBO;
import org.gorson.function.ProcessTransformSensorTimestampToStringFunction;
import org.gorson.source.SensorSource1;
import org.gorson.source.SensorSource2;
import java.time.Duration;

public class SensorJoinDemo {

    public static void main(String[] args) throws Exception {
        //使用本地模式并开启WebUI
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT, "8081-8082");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        env.setParallelism(1);

        SingleOutputStreamOperator<SensorReadingBO> sensorSource1 = env.addSource(new SensorSource1())
                .assignTimestampsAndWatermarks(
/*                        new AscendingTimestampExtractor<SensorReadingBO>() {
                            @Override
                            public long extractAscendingTimestamp(SensorReadingBO sensorReadingBO) {
                                return sensorReadingBO.timestamp;
                            }
                        });*/
                        WatermarkStrategy
                                .<SensorReadingBO>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((event, timestamp) -> event.timestamp));

        SingleOutputStreamOperator<SensorReadingBO> sensorSource2 = env.addSource(new SensorSource2())
                .assignTimestampsAndWatermarks(
/*                        new AscendingTimestampExtractor<SensorReadingBO>() {
                            @Override
                            public long extractAscendingTimestamp(SensorReadingBO sensorReadingBO) {
                                return sensorReadingBO.timestamp;
                            }
                        });*/
                        WatermarkStrategy
                                .<SensorReadingBO>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((event, timestamp) -> event.timestamp));

        sensorSource1.print("ps1 ");
        sensorSource2.print("ps2 ");

        SingleOutputStreamOperator<Tuple3<String, String, Double>> processSource1 = sensorSource1.process(new ProcessTransformSensorTimestampToStringFunction());
        SingleOutputStreamOperator<Tuple3<String, String, Double>> processSource2 = sensorSource2.process(new ProcessTransformSensorTimestampToStringFunction());

        processSource1.print("ps1 ");
        processSource2.print("ps2 ");

        DataStream<Tuple5<String, String, Double, Double, Double>> joinSource = processSource1.setParallelism(1)
                .join(processSource2)
                .where(new KeySelector<Tuple3<String, String, Double>, String>() {
                    @Override
                    public String getKey(Tuple3<String, String, Double> tuple3) throws Exception {
                        return tuple3.f1;
                    }
                })
                .equalTo(new KeySelector<Tuple3<String, String, Double>, String>() {
                    @Override
                    public String getKey(Tuple3<String, String, Double> tuple3) throws Exception {
                        return tuple3.f1;
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(1)))
                .apply(new JoinFunction<Tuple3<String, String, Double>, Tuple3<String, String, Double>, Tuple5<String, String, Double, Double, Double>>() {
                    @Override
                    public Tuple5<String, String, Double, Double, Double> join(Tuple3<String, String, Double> source1, Tuple3<String, String, Double> source2) throws Exception {
                        return Tuple5.of(
                            source1.f0 + "-" + source2.f0, //id
                            source1.f1, //timestamp
                            source1.f2, // source1 temperature
                            source2.f2, // source2 temperature
                            source1.f2 + source2.f2
                        );
                    }
                });

        joinSource.print("join =");
        env.execute("");
    }

}
